/* Copyright (c) 2023 Renmin University of China
RMDB is licensed under Mulan PSL v2.
You can use this software according to the terms and conditions of the Mulan PSL v2.
You may obtain a copy of Mulan PSL v2 at:
        http://license.coscl.org.cn/MulanPSL2
THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
See the Mulan PSL v2 for more details. */

#pragma once

#include "execution_defs.h"
#include "execution_manager.h"
#include "executor_abstract.h"
#include "index/ix.h"
#include "system/sm.h"

class IndexScanExecutor : public AbstractExecutor {
   private:
    std::string tab_name_;                      // 表名称
    TabMeta tab_;                               // 表的元数据
    std::vector<Condition> conds_;              // 扫描条件
    RmFileHandle *fh_;                          // 表的数据文件句柄
    std::vector<ColMeta> cols_;                 // 需要读取的字段
    size_t len_;                                // 选取出来的一条记录的长度
    std::vector<Condition> fed_conds_;          // 扫描条件，和conds_字段相同

    std::vector<std::string> index_col_names_;  // index scan涉及到的索引包含的字段
    IndexMeta index_meta_;                      // index scan涉及到的索引元数据

    Rid rid_;
    std::unique_ptr<RecScan> scan_;
    std::unique_ptr<RmRecord> rec_;

    SmManager *sm_manager_;
    bool isend_;

   public:
    IndexScanExecutor(SmManager *sm_manager, std::string tab_name, std::vector<Condition> conds, std::vector<std::string> index_col_names,
                    Context *context) {
        sm_manager_ = sm_manager;
        context_ = context;
        tab_name_ = std::move(tab_name);
        tab_ = sm_manager_->db_.get_table(tab_name_);
        conds_ = std::move(conds);
        // index_no_ = index_no;
        index_col_names_ = index_col_names; 
        index_meta_ = *(tab_.get_index_meta(index_col_names_));
        fh_ = sm_manager_->fhs_.at(tab_name_).get();
        cols_ = tab_.cols;
        len_ = cols_.back().offset + cols_.back().len;
        std::map<CompOp, CompOp> swap_op = {
            {OP_EQ, OP_EQ}, {OP_NE, OP_NE}, {OP_LT, OP_GT}, {OP_GT, OP_LT}, {OP_LE, OP_GE}, {OP_GE, OP_LE},
        };

        for (auto &cond : conds_) {
            if (cond.lhs_col.tab_name != tab_name_) {
                // lhs is on other table, now rhs must be on this table
                assert(!cond.is_rhs_val && cond.rhs_col.tab_name == tab_name_);
                // swap lhs and rhs
                std::swap(cond.lhs_col, cond.rhs_col);
                cond.op = swap_op.at(cond.op);
            }
        }
        fed_conds_ = conds_;
        isend_ = false;
        // 先获取对应的表锁
        // context_->lock_mgr_->lock_shared_on_table(context_->txn_.get(), fh_->GetFd());
    }

    void beginTuple() override {
        isend_ = false;
        // 先获取对应的表锁
        // context_->lock_mgr_->lock_IS_on_table(context_->txn_.get(), fh_->GetFd());
        // index is available, scan index
        auto ih = sm_manager_->ihs_.at(index_meta_.index_name).get();
        // 根据条件调整scan_ 范围[lower, upper)
        bool is_set_lower = false;
        bool is_set_upper = false;
        std::unique_ptr<char[]> lower_key = index_meta_.get_min_key();
        std::unique_ptr<char[]> upper_key = index_meta_.get_max_key();
        for (int i = 0; i < index_meta_.col_num; i ++){
            for(auto &cond : fed_conds_) {
                if (cond.is_rhs_val && cond.op != OP_NE && cond.lhs_col.col_name == index_meta_.cols[i].name) {
                    if (cond.op < OP_EQ || cond.op > OP_GE) {
                        throw InternalError("IndexScanExecutor: Unexpected op type");
                    }
                    if (cond.op == OP_EQ || cond.op == OP_GT || cond.op == OP_GE) {
                        if (i == 0) is_set_lower = true;
                        index_meta_.set_key(lower_key.get(), cond.lhs_col.col_name, cond.rhs_val);
                    }
                    if (cond.op == OP_EQ || cond.op == OP_LT || cond.op == OP_LE) {
                        if (i == 0) is_set_upper = true;
                        index_meta_.set_key(upper_key.get(), cond.lhs_col.col_name, cond.rhs_val);
                    }
                }
            }
        }
        // TODO: 这部分后期移动到逻辑优化部分
        // 判断lower和upper是否合法
        // leaf_begin() 使用到first_leaf_，这个变量没修改过。
        // Iid lower = is_set_lower ? ih->lower_bound(lower_key.get()) : ih->leaf_begin();
        Iid lower = ih->lower_bound(lower_key.get());
        Iid upper = is_set_upper ? ih->upper_bound(upper_key.get()) : ih->leaf_end();
        if (lower == ih->leaf_end()) {
            isend_ = true;
            return;
        }
        if (is_set_lower && is_set_upper) {
            if (ix_compare(lower_key.get(), upper_key.get(), index_meta_.cols) > 0) {
                isend_ = true;
                return;
            }
        }
        
        scan_ = std::make_unique<IxScan>(ih, lower, upper, sm_manager_->get_bpm());
        while (!scan_->is_end()) {
            rid_ = scan_->rid();
            try {
                rec_ = fh_->get_record(rid_, context_);
                if (eval_conds(cols_, fed_conds_, rec_.get())) {
                    break;
                }
            } catch (RecordNotFoundError &e) {
                
            }
            scan_->next();
        } 
    }

    void nextTuple() override {
        for (scan_->next(); !scan_->is_end(); scan_->next()) {
            rid_ = scan_->rid();
            rec_ = fh_->get_record(rid_, context_);
            if (eval_conds(cols_, fed_conds_, rec_.get())) return;
        }
    }

    std::unique_ptr<RmRecord> Next() override {
        assert(!is_end());
        return std::move(rec_);
    }

    bool is_end() const override {
        if (isend_) return true;
        return scan_->is_end();
    }

    size_t tupleLen() const override { return len_; }

    const std::vector<ColMeta> &cols() const override { return cols_; }

    Rid &rid() override { return rid_; }

    /**
     * @brief 判断记录 rec 是否满足条件 cond 
     * 
     * @param rec_cols 记录 rec 的列名等信息
     * @param cond 查询条件
     * @param rec 
     * @return true 
     * @return false 
     */
    bool eval_cond(const std::vector<ColMeta> &rec_cols, const Condition &cond, const RmRecord *rec) {
        auto lhs_col = get_col(rec_cols, cond.lhs_col);
        char *lhs = rec->data + lhs_col->offset;
        char *rhs;
        ColType rhs_type;
        if (cond.is_rhs_val) {
            rhs_type = cond.rhs_val.type;
            rhs = cond.rhs_val.raw->data;
        } else {
            // rhs is a column
            // rhs 所在表是当前表
            auto rhs_col = get_col(rec_cols, cond.rhs_col);
            rhs_type = rhs_col->type;
            rhs = rec->data + rhs_col->offset;
        }
        int cmp = ix_compare(lhs, rhs, lhs_col->type, rhs_type, lhs_col->len);
        if (cond.op == OP_EQ) {
            return cmp == 0;
        } else if (cond.op == OP_NE) {
            return cmp != 0;
        } else if (cond.op == OP_LT) {
            return cmp < 0;
        } else if (cond.op == OP_GT) {
            return cmp > 0;
        } else if (cond.op == OP_LE) {
            return cmp <= 0;
        } else if (cond.op == OP_GE) {
            return cmp >= 0;
        } else {
            throw InternalError("Unexpected op type");
        }
    }


    bool eval_conds(const std::vector<ColMeta> &rec_cols, const std::vector<Condition> &conds, const RmRecord *rec) {
        return std::all_of(conds.begin(), conds.end(),
                           [&](const Condition &cond) { return eval_cond(rec_cols, cond, rec); });
    }
};